#include <sys/types.h>
#include <unistd.h>
#include <sys/ioctl.h>
#include <sstream>
#include <string>
#include <stdlib.h>
#include <sys/stat.h>
#include <libpq-fe.h>
#include <map>
#include <glib.h>
#include <string.h>
#include "config.h"
#include "Debug.h"
#include "define.h"

using namespace std;

extern "C" {
  bool msnqqConsInit(char *consumer, const char* conf);
  bool msnqqConsFriends(char *user, GHashTable *frs);
  void msnqqConsShutdown();
  void msnqqConsFlush(char *consumer);
};

PGconn *g_db = NULL;

bool msnqqConsInit(char *consumer, const char* conf)
{
        struct stat st;
        if (lstat(conf, &st) != 0)
        {
                ERRO<<":"<<conf<<" is not exist!"<<endl;
                return false;
        }
        Configure cfg;
        map <string, string > g_Confs;
        g_Confs = cfg.LoadConfInfo(conf);
        cfg.Print();
        if(g_Confs.size() <= 0)
        {
                cerr<<"error to explain conf :"<< conf<<endl;
                return true;
        }
        string sval = "";
        map <string, string >::iterator iter;

        string dbhost,  dbuser, dbpass, dbname;
        if (g_Confs.find("dbhost") != g_Confs.end())
                dbhost = g_Confs["dbhost"];

        if (g_Confs.find("dbuser") != g_Confs.end())
                dbuser = g_Confs["dbuser"];
        if (g_Confs.find("dbpass") != g_Confs.end())
                dbpass = g_Confs["dbpass"];
        if (g_Confs.find("dbname") != g_Confs.end())
		dbname = g_Confs["dbname"];
        string dbport = "5432";
        if (g_Confs.find("dbport") != g_Confs.end())
		dbport = g_Confs["dbport"];
	g_db = PQsetdbLogin(dbhost.c_str(), dbport.c_str(), NULL, NULL, dbname.c_str(), dbuser.c_str(), dbpass.c_str());

	if(PQstatus(g_db) != CONNECTION_OK)
	{
		ERRO << PQerrorMessage(g_db) << endl;
		PQfinish( g_db);
		g_db = NULL;
                return false;
	}

        return true;
}

bool msnqqConsFriends(char * user, GHashTable *frs)
{
	if(user != NULL && g_db != NULL)
	{
		char uval[ strlen(user)*2+ 1 ];
		int elen = PQescapeString(uval, user, strlen(user));
		uval[elen] = '\0';

		stringstream sql;

		sql<<"SELECT nid, type, list FROM user_import WHERE source ='"<< uval <<"' LIMIT 100" ;
		PGresult *result = PQexec(g_db, sql.str().c_str());
		if(PQresultStatus(result) != PGRES_TUPLES_OK)
		{
			cerr << PQerrorMessage(g_db) << endl;
			PQclear(result);
			return false;
		}

		long rownum = PQntuples(result);
		if (rownum == 0 )
		{
			PQclear(result);
			return true;
		}
		string nids = ""; 
		for (int i = 0; i < rownum; i++)
		{
			if(i != 0)
				nids.append(";");
			nids.append((char *)PQgetvalue(result, i , 0));
		}

                string subwhere;
		for (int i = 0; i < rownum; i++)
		{
                        int itype = atoi(PQgetvalue(result, i, 1));
			char **columns = g_strsplit( PQgetvalue(result, i, 2), ";", -1);
			int columns_num =  g_strv_length(columns);
			for(int i=0; i< columns_num ; i++ )
			{ 
				char *nuser = g_strdup(g_strstrip(columns[i]));
                                if(subwhere.length() > 0)
                                     subwhere.append(",");
                                subwhere.append("'");
                                subwhere.append(nuser);
                                subwhere.append("'"); 
				FObject *uself = new FObject(nuser, itype ,(char *)"");
				g_hash_table_insert(frs, nuser, uself);
			}
			g_strfreev(columns);
                }
		PQclear(result);

//select nid from user_import where source in('hsg1158@msn.cn', '68826862@qq.com');
                sql.str("");
		sql<<"SELECT source, nid FROM user_import WHERE source IN("<< subwhere.c_str()<<")";
		result = PQexec(g_db, sql.str().c_str());
		if(PQresultStatus(result) != PGRES_TUPLES_OK)
		{
			cerr << PQerrorMessage(g_db) << endl;
			PQclear(result);
			return false;
		}

		rownum = PQntuples(result);
		for (int i = 0; i < rownum; i++)
		{
                       
 			FObject *user = (FObject *)g_hash_table_lookup(frs, PQgetvalue(result, i , 0));
                        if(user != NULL)
                            user->value = PQgetvalue(result, i, 1);
		}
		PQclear(result);
	}
	return true;
}

void msnqqConsShutdown()
{
	if(g_db != NULL){
		PQfinish( g_db);
		g_db = NULL;
	} 

 // Do whatever to close/flush/sync the consumer's work before exit...
  //   delete gx_TxLog;
}

void msnqqConsFlush(char * consumer)
{
  // If this consumer keeps statistics or writes to a tx log, then
  // flush the statistics here, otherwise do nothing.
}


